package com.inspinia.base.mq;

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

/**
 * Created by veblen on 16/5/3.
 */
public class KafkaConsumerHelper {
    private static final Logger logger = LoggerFactory.getLogger(KafkaProducerHelper.class);


    private static Properties loadProperties(String groupId) {
        //加载配置文件
        Properties props = new Properties();

        props.put("bootstrap.servers",KafkaConfig.BOOTSTRAP_SERVERS);
        props.put("enable.auto.commit", KafkaConfig.ENABLE_AUTO_COMMIT);
        props.put("session.timeout.ms", KafkaConfig.SESSION_TIMEOUT_MS);
        props.put("auto.offset.reset", KafkaConfig.AUTO_OFFSET_RESET);
        props.put("key.deserializer", KafkaConfig.KEY_DESERIALIZER);
        props.put("value.deserializer", KafkaConfig.VALUE_DESERIALIZER);
        return props;
    }

    /**
     * 创建container
     *
     * @param groupId 消费者group
     * @param topics  订阅的MQ的类目
     * @return
     */
    private static KafkaConsumer<String, String> createContainer(String groupId, String... topics) {
        Properties properties = loadProperties(groupId);
        return new KafkaConsumer<>(properties);
    }

    /**
     * 获取消费者类
     *
     * @param groupId 消费者group
     * @param topics  订阅的MQ的类目
     * @return
     */
    public static KafkaConsumer<String, String> getConsumer(String groupId, String... topics) {

        return createContainer(groupId, topics);
    }


}
